package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/24 11:39
 */
public class Dwd_15_DwdInteractionCommentApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_15_DwdInteractionCommentApp().init(2015, 2, "Dwd_15_DwdInteractionCommentApp");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        
        readOdsDb(tEnv, "Dwd_15_DwdInteractionCommentApp");
        
        readBaseDic(tEnv);
    
        Table commentInfo = tEnv.sqlQuery("select " +
                                        "data['id'] id, " +
                                        "data['user_id'] user_id, " +
                                        "data['sku_id'] sku_id, " +
                                        "data['order_id'] order_id, " +
                                        "data['create_time'] create_time, " +
                                        "data['appraise'] appraise, " +
                                        "pt, " +
                                        "ts " +
                                        "from ods_db " +
                                        "where `table`='comment_info' " +
                                        "and `type`='insert' ");
        tEnv.createTemporaryView("comment_info", commentInfo);
    
    
        Table resultTable = tEnv.sqlQuery("select " +
                                        "ci.id, " +
                                        "ci.user_id, " +
                                        "ci.sku_id, " +
                                        "ci.order_id, " +
                                        "date_format(ci.create_time,'yyyy-MM-dd') date_id, " +
                                        "ci.create_time, " +
                                        "ci.appraise appraise_code, " +
                                        "dic.dic_name appraise_name, " +
                                        "ts " +
                                        "from comment_info ci " +
                                        "join base_dic for system_time as of ci.pt dic on ci.appraise=dic.dic_code ");
    
        tEnv.executeSql("create table dwd_interaction_comment( " +
                                "id string, " +
                                "user_id string, " +
                                "sku_id string, " +
                                "order_id string, " +
                                "date_id string, " +
                                "create_time string, " +
                                "appraise_code string, " +
                                "appraise_name string, " +
                                "ts string " +
                                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_INTERACTION_COMMENT));
        
        resultTable.executeInsert("dwd_interaction_comment");
    
    
    
    }
}
